package com.bw.app.dwd;


import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//dwd层进行字段的筛选 减少数据量，选择需要的字段

public class Dwd_gd12 {
    public static void main(String[] args) throws Exception {
//        环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

//        读取日志表
        tEnv.executeSql("create table ods_behavior_log(" +
                "page_id string," +
                "user_id string," +
                "product_id string," +
                "behavior_type string," +
                "behavior_time string" +
                ")" + MyKafkaUtil.getKafkaDDL("gd12_ods_behavior_log_topic", "a"));

//        tEnv.executeSql("select * from dwd_behavior_log").print();


//        订单表
        tEnv.executeSql("create table ods_order(" +
                "order_id string," +
                "user_id string," +
                "product_id string," +
                "order_amount string," +
                "order_time string," +
                "order_status string" +
                ")" + MyKafkaUtil.getKafkaDDL("gd12_ods_order_topic", "b"));

//        商品表
        tEnv.executeSql("create table ods_product(" +
                "product_id string," +   //商品id
                "product_title string," +  //商品名称
                "product_category string," +  //商品分类
                "product_price string," +  //价格
                "product_description string" +  //商品描述
                ")" + MyKafkaUtil.getKafkaDDL("gd12_ods_product_topic", "c"));

//        用户表
        tEnv.executeSql("create table ods_user(" +
                "user_id string," +
                "user_name string," +
                "age string," +
                " gender string," +
                " region string," +
                " registration_time string " +
                ")" + MyKafkaUtil.getKafkaDDL("gd12_ods_user_topic", "d"));



//        用户表join日志表join商品表
        Table dwd_join = tEnv.sqlQuery("SELECT  " +
                "ou.user_id, " +
                "op.product_title, " +
                "ol.behavior_type, " +
                "ol.behavior_time  " +
                " FROM ods_user ou " +
                " JOIN ods_behavior_log ol ON ou.user_id = ol.user_id " +
                " JOIN ods_product op on ol.product_id = op.product_id");

        tEnv.createTemporaryView("dwd_join",dwd_join);


        tEnv.executeSql("create table dwd_user_join( " +
                "user_id string, " +
                "product_title string, " +
                "behavior_type string, " +
                "behavior_time string " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_user_join_topic"));


        tEnv.executeSql("insert into dwd_user_join select * from dwd_join");

//        tEnv.executeSql("select count(*) from dwd_user_join").print();
//        tEnv.executeSql("select behavior_type,count(*) from dwd_user_join GROUP BY behavior_type").print();

//        env.execute();





    }
}
